package com.chengqs.dfs.namenode.server;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <B>主类名称：</B>DataNodeManager<BR>
 * <B>概要说明：</B>这个组件，就是负责管理集群里的所有的datanode的<BR>
 *
 * @author: chengqs
 * @create: 2025-03-30 03:47:24
 **/
@Slf4j
public class DataNodeManager {
    /**
     * 内存中维护的datanode列表
     */
    private Map<String, DataNodeInfo> dataNodes = new ConcurrentHashMap<>();

    public DataNodeManager() {
        new DataNodeAliveMonitor().start();
    }

    /**
     * datanode进行注册
     * @param ip
     * @param hostname
     */
    public Boolean register(String ip, String hostname) {
        DataNodeInfo dataNode = new DataNodeInfo(ip, hostname);
        dataNodes.put(ip + "-" + hostname, dataNode);
        log.info("dataNode注册：ip = {},hostname = {}", ip, hostname);
        return true;
    }

    /**
     * datanode进行心跳
     * @param ip
     * @param hostname
     * @return
     */
    public Boolean heartbeat(String ip, String hostname) {
        DataNodeInfo dataNode = dataNodes.get(ip + "-" + hostname);
        dataNode.setLatestHeartbeatTime(System.currentTimeMillis());
        log.info("dataNode发送心跳：ip = {},hostname = {}", ip, hostname);
        return true;
    }

    /**
     * datanode是否存活的监控线程
     */
    class DataNodeAliveMonitor extends Thread {
        @Override
        public void run() {
            try {
                while (true) {
                    List<String> toRemoveDataNodes = new ArrayList<>();

                    for (DataNodeInfo dataNode : dataNodes.values()) {
                        if (System.currentTimeMillis() - dataNode.getLatestHeartbeatTime() > 90 * 1000) {
                            toRemoveDataNodes.add(dataNode.getIp() + "-" + dataNode.getHostname());
                        }
                    }

                    if (!toRemoveDataNodes.isEmpty()) {
                        for (String toRemoveDataNode : toRemoveDataNodes) {
                            dataNodes.remove(toRemoveDataNode);
                        }
                    }

                    Thread.sleep(30 * 1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
